package com.nokia.position;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import com.nokia.position.PositionProto.Position;
import com.twitter.elephantbird.mapreduce.input.MultiInputFormat;
import com.twitter.elephantbird.mapreduce.io.ProtobufWritable;
import com.twitter.elephantbird.mapreduce.output.LzoProtobufB64LineOutputFormat;
import com.twitter.elephantbird.mapreduce.output.LzoProtobufBlockOutputFormat;

public class PositionPartitionDriver {
	
	public void printUsage() {
			System.out
					.println("Usage: hadoop jar path/to/this.jar "
							+ this.getClass()
							+ " <input dir> <output dir>");

	}

	public int run(String[] args, Configuration conf)
			throws Exception {
		
		if (args.length != 2) {
			printUsage();
			return 1;
		}
		
        Job job = new Job(conf);
        job.setJobName("Positioning Partition");

        job.setJarByClass(getClass());
        job.setMapperClass(PositionPartitionMapper.class);
        job.setReducerClass(PositionPartitionReducer.class);

        // input format is same for both B64Line or block:
        MultiInputFormat.setInputFormatClass(Position.class, job);
        job.setMapOutputKeyClass(LongWritable.class);
        job.setMapOutputValueClass(ProtobufWritable.class);

        job.setOutputKeyClass(LongWritable.class);
        job.setOutputValueClass(ProtobufWritable.class);

        if (conf.get("proto.test.format", "B64Line").equals("Block")) {
                LzoProtobufBlockOutputFormat.setClassConf(Position.class,
                                job.getConfiguration());
                job.setOutputFormatClass(LzoProtobufBlockOutputFormat.class);
        } else { // assume B64Line
                LzoProtobufB64LineOutputFormat.setClassConf(Position.class,
                                job.getConfiguration());
                job.setOutputFormatClass(LzoProtobufB64LineOutputFormat.class);
        }

        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        return job.waitForCompletion(true) ? 0 : 1;
	}
	
	public static void main(String[] args) throws Exception {
		
		Configuration conf = new Configuration();
		args = new GenericOptionsParser(conf, args).getRemainingArgs();
		PositionPartitionDriver driver = new PositionPartitionDriver();
		System.exit(driver.run(args, conf));
	}
}